package cn.monkey.io.transport.mq.redisson;

import cn.monkey.io.transport.mq.MqInboundConfig;
import org.redisson.api.RFuture;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;

import java.util.List;
import java.util.function.Predicate;
import java.util.stream.Stream;

public class SimpleRedissonInbound implements RedissonInbound {
    private final MqInboundConfig inboundConfig;
    private final RedissonFluxReceive receiver;

    public SimpleRedissonInbound(RedissonClient redissonClient,
                                 Codec codec,
                                 MqInboundConfig inboundConfig) {
        this.inboundConfig = inboundConfig;
        List<RTopic> topicList = Stream.of(inboundConfig.getTopics())
                .map(topic -> redissonClient.getTopic(topic, codec))
                .toList();
        this.receiver = new RedissonFluxReceive(topicList);
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver.filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }

    @Override
    public Mono<Void> close() {
        this.receiver.dispose();
        return Mono.empty();
    }

    @Override
    public MqInboundConfig inboundConfig() {
        return this.inboundConfig;
    }
}
